package spark.core.java;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

/**
 * 分组取TOP3 取每个班级最大3个分数
 */
public class GroupTop3 {

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("SortWordCount").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        final JavaRDD<String> lines = sc.textFile("datas/groupTop.txt");
        JavaPairRDD<String,Integer> pairs = lines.mapToPair(new PairFunction<String, String, Integer>() {
            public Tuple2<String, Integer> call(String line) throws Exception {
                String[] lineSplited = line.split(" ");
                return new Tuple2<String, Integer>(lineSplited[0], Integer.valueOf(lineSplited[1]));
            }
        });
        JavaPairRDD<String,Iterable<Integer>> groupedPairs =  pairs.groupByKey();
        JavaPairRDD<String,Iterable<Integer>> top3Score = groupedPairs.mapToPair(
                new PairFunction<Tuple2<String, Iterable<Integer>>, String, Iterable<Integer>>() {
                    public Tuple2<String, Iterable<Integer>> call(Tuple2<String, Iterable<Integer>> classscores) throws Exception {
                        Integer[] top3 = new Integer[3];
                        String className = classscores._1();
                        Iterator<Integer> scores = classscores._2().iterator();
                        while (scores.hasNext()) {
                            Integer score = scores.next();
                            for (int i = 0; i < 3; i++) {
                                if (top3[i] == null) {
                                    top3[i] = score;
                                    break;
                                } else if (score > top3[i]) {
                                    for(int j= 2;j>1;j--){
                                        top3[j] = top3[j-1];
                                    }
                                    top3[i] = score;
                                    break;
                                }
                            }
                        }
                        return new Tuple2<String, Iterable<Integer>>(className, Arrays.asList(top3));
                    }
                });

        top3Score.foreach(new VoidFunction<Tuple2<String, Iterable<Integer>>>() {
            public void call(Tuple2<String, Iterable<Integer>> v) throws Exception {
                System.out.println(v._1());
                Iterator<Integer> ite = v._2().iterator();
                while(ite.hasNext()){
                    System.out.println(ite.next());
                }
            }
        });

    }
}
